package cn.intotw.rdcj.core.compoment;

import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.util.MimeTypeUtils;

/**
 * @author Chenxiang
 * @Description: RocketMQ消息发送
 * @create 2020/4/29 16:55
 */

@Slf4j
public class RocketMqCompoment{
    RocketMQTemplate rocketMQTemplate;
    Gson gson;

    public RocketMqCompoment(RocketMQTemplate rocketMQTemplate, Gson gson) {
        this.rocketMQTemplate = rocketMQTemplate;
        this.gson = gson;
    }

    DefaultCallBack defaultCallBack=new DefaultCallBack();
    public SendResult syncSend(String topic,String tag, String message,long timeOut, int delayLevel) {
        return rocketMQTemplate.syncSend(makeDestination(topic,tag), buildMessage(message), timeOut, delayLevel);
    }
    public SendResult syncSend(String topic,String tag, String message,long timeOut) {
        return rocketMQTemplate.syncSend(makeDestination(topic,tag), buildMessage(message), timeOut);
    }
    public void asyncSend(String topic,String tag, String message,long timeOut) {
        rocketMQTemplate.asyncSend(makeDestination(topic,tag), buildMessage(message),defaultCallBack,timeOut);
    }
    public void asyncSend(String topic,String tag, String message,long timeOut,int delayLevel) {
        rocketMQTemplate.asyncSend(makeDestination(topic,tag), buildMessage(message),defaultCallBack, timeOut,delayLevel);
    }
    public void asyncSend(String topic,String tag, String message,SendCallback sendCallback,long timeOut,int delayLevel) {
        rocketMQTemplate.asyncSend(makeDestination(topic,tag), buildMessage(message),sendCallback, timeOut,delayLevel);
    }
    public void asyncSend(String topic,String tag, String message,SendCallback sendCallback,long timeOut) {
        rocketMQTemplate.asyncSend(makeDestination(topic,tag), buildMessage(message),sendCallback, timeOut);
    }
    private String makeDestination(String topic,String tag){
        return topic.concat(":").concat(tag);
    }
    private Message buildMessage(Object message){
        if(message instanceof String){
            return MessageBuilder.withPayload(
                    message).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE).build();
        }else {
            return MessageBuilder.withPayload(
                    gson.toJson(message)).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE).build();
        }
    }
    private static class DefaultCallBack implements SendCallback{
        @Override
        public void onSuccess(SendResult sendResult) {
            log.info("异步发送mq消息成功，msgId={}",sendResult.getMsgId());
        }

        @Override
        public void onException(Throwable throwable) {
            log.error("异步发送mq消息失败",throwable);
        }
    }
}